package com.hliushi.spark.sql

import org.apache.spark.sql.types._
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/18 14:34
 */
object HiveWrite {


  /**
   * 读取HDFS上面的文件数据, 然后将其写入到Hive表中
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {
    // 1.创建SparkSession
    //    1.1.开启Hive支持            经过这个配置, SparkSQL才会把SQL语句当做HiveSQL俩进行解析
    //    1.2.指定MetaStore的位置
    //    1.3.指定warehouse的位置
    /**
     * 提交到集群上运行, 这个 .master() 参数不能配置
     */
    val spark = SparkSession.builder()
      .master("local[6]") // 如果在集群上运行[spark-submit, 一定要注释掉]
      .appName("hive_write")
      .enableHiveSupport() // 1.开启Hive支持
      .config("hive.metastore.uris", "thrift://node03:9083") // 2.指定MetaStore的位置
      //.config("spark.sql.warehouse.dir", "/dataset/hive")
      .config("spark.sql.warehouse.dir", "hdfs://node01:8020/dataset/hive") // 3.指定warehouse的位置
      .getOrCreate()


    // 2.读取数据 [读取hdfs上的文件]
    //    2.1.上传HDFS, 因为要在集群中执行, 没办法保证程序在哪个文件机器中执行
    //      要想读取本地文件, 需要把文件上传到所有的机器中
    //      上传到HDFS中就可以解决这个问题, 所有的机器都可以读取到HDFS中的文件
    //      它是一个外部系统
    //    2.2.使用DF读取数据
    //       hdfs:///dataset/studenttab10k
    val schema = StructType(
      List(
        StructField("name", StringType),
        StructField("age", IntegerType),
        StructField("gpa", FloatType)
      )
    )
    // 0.导入隐式转换
    import spark.implicits._

    val dataFrame = spark.read
      .option("delimiter", "\t")
      .schema(schema)
      .csv("hdfs://node01:8020/dataset/studenttab10k")


    val resultDF = dataFrame.where($"age" > 50)

    // 3.写入到hive中, 使用写入表的API, saveAsTable
    resultDF.write.mode(SaveMode.Overwrite)
      .saveAsTable("spark01.stu04")

    spark.stop()
  }
}